package com.thp.bigdata.mapSideJoin;

import java.io.BufferedReader;
import java.io.FileInputStream;
import java.io.IOException;
import java.io.InputStreamReader;
import java.net.URI;
import java.net.URISyntaxException;
import java.util.HashMap;
import java.util.Map;

import org.apache.commons.lang.StringUtils;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

/**
 * 在Map端实现join，不需要在join端实现join，这个可以减少数据倾斜
 * 适用于关联表中有小表的情况
 * 可以将小表分发到所有的map节点，这样，map节点就可以在本地对自己所读到的大表数据进行join并输出最终结果
 * 可以大大提高join操作的并发操作，加快处理速度
 * 
 * 【博客链接】
 * 		http://dongxicheng.org/mapreduce-nextgen/hadoop-distributedcache-details/
 * 		https://blog.csdn.net/xiaolang85/article/details/11782539
 * 
 * @author 汤小萌
 *
 */
public class MapSideJoin {
	
	static class MapSideJoinMapper extends Mapper<LongWritable, Text, Text, NullWritable> {
		// 使用一个HashMap来加载保存产品信息表
		Map<String, String> pdInfoMap = new HashMap<String, String>();
		
		Text k = new Text();
		
		// 在maptask处理数据之前调用一次，可以用来做一些初始化的操作
		@Override
		protected void setup(Context context)
				throws IOException, InterruptedException {
			// 已经加载到当前 maptask的工作目录下了，可以直接写文件名读取文件
			// hadoop 所有读取数据都是使用的utf-8
			BufferedReader br = new BufferedReader(new InputStreamReader(new FileInputStream("pdts.txt")));
			String line;
			while(StringUtils.isNotEmpty(line = br.readLine())) {
				String[] fields = line.split("\t");
				pdInfoMap.put(fields[0], fields[1]);
			}
			br.close();
		}
		
		// 由于已经持有完整的产品信息表，所以可以直接在map方法中就能实现join的逻辑
		@Override
		protected void map(LongWritable key, Text value, Context context)
				throws IOException, InterruptedException {
			String orderLine = value.toString();
			String[] fields = orderLine.split("\t");
			String pdName = pdInfoMap.get(fields[1]);
			k.set(orderLine + "\t" + pdName);
			context.write(k, NullWritable.get());
		}
		
		
		public static void main(String[] args) throws IOException, URISyntaxException, ClassNotFoundException, InterruptedException {
			Configuration conf = new Configuration();
			
			Job job = Job.getInstance(conf);
			
			
			job.setJarByClass(MapSideJoin.class);
			
			job.setMapperClass(MapSideJoinMapper.class);
			
			job.setOutputKeyClass(Text.class);
			job.setOutputValueClass(NullWritable.class);
			
			
			FileInputFormat.setInputPaths(job, new Path("F:/mapSideJoin/input"));
			FileOutputFormat.setOutputPath(job, new Path("F:/mapSideJoin/output"));
			
			// 指定需要缓存一个文件到所有的maptask运行节点工作目录
			/* job.addArchiveToClassPath(archive); */// 缓存jar包到task运行节点的classpath中
			/* job.addFileToClassPath(file); */// 缓存普通文件到task运行节点的classpath中
			/* job.addCacheArchive(uri); */// 缓存压缩包文件到task运行节点的工作目录
			/* job.addCacheFile(uri) */// 缓存普通文件到task运行节点的工作目录
			
			// 将产品表文件缓存到task工作节点的工作目录中去
			job.addCacheFile(new URI("file:/F:/mapSideJoin/pdts.txt"));
			
			// map端的join的逻辑不需要reduce阶段，设置reducetask数量为0，不设置是会出现问题的
			job.setNumReduceTasks(0);
			
			boolean res = job.waitForCompletion(true);
			
			System.exit(res ? 0 : 1);
			
		}
		
	}
	
	
}
